package SparkStreaming;

import com.google.protobuf.GeneratedMessage;
import com.googlecode.protobuf.format.JsonFormat;
import com.sdyc.ndmp.protobuf.hadoop.ProtobufInputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.io.Serializable;

/**
 * <pre>
 * Created with IntelliJ IDEA.
 * User: zhengzhi
 * Date: 2017/5/11
 * To change this template use File | Settings | File Templates.
 * </pre>
 */
public class ProtobufSourceInput implements Serializable {

    private transient JavaSparkContext sparkContext;
    private transient Configuration cfg;

    public ProtobufSourceInput(JavaSparkContext sparkContext, Configuration cfg) {
        this.sparkContext = sparkContext;
        this.cfg = cfg;
    }

    public JavaRDD<String> getSourceRDD() {
        return sparkContext.newAPIHadoopRDD(cfg, ProtobufInputFormat.class, LongWritable.class, GeneratedMessage.class)
                .map(new Function<Tuple2<LongWritable, GeneratedMessage>, String>() {
                    @Override
                    public String call(Tuple2<LongWritable, GeneratedMessage> value) throws Exception {
                        return JsonFormat.printToString(value._2());
                    }
                });
    }

//hdfs://nowledgedata-n7:9000/data/tracker/20170511/12-data-0,hdfs://nowledgedata-n7:9000/data/tracker/20170511/12-data-1
    public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local[2]", "test");
        Configuration conf = new Configuration();

        String inputpath = "hdfs://nowledgedata-n7:9000/data/tracker/20170511/17-data-0";
        conf.set("output.protobuf.class", UserAgent.UserAgentDetail.class.getName());
        conf.set("mapreduce.input.fileinputformat.inputdir",inputpath);


        ProtobufSourceInput input = new ProtobufSourceInput(sc , conf);

        input.getSourceRDD().foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }
}
